package com.example.oauth.tracing.reporter;

import com.example.oauth.utils.SpELUtil;
import io.jaegertracing.internal.JaegerSpan;
import io.jaegertracing.internal.exceptions.SenderException;
import io.jaegertracing.internal.metrics.InMemoryMetricsFactory;
import io.jaegertracing.internal.metrics.Metrics;
import io.jaegertracing.internal.senders.SenderResolver;
import io.jaegertracing.spi.Reporter;
import io.jaegertracing.spi.Sender;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;

import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

@ToString
@Slf4j
public class EnhancedRemoteReporter implements Reporter {
    private static final int DEFAULT_CLOSE_ENQUEUE_TIMEOUT_MILLIS = 1000;

    public static final int DEFAULT_FLUSH_INTERVAL_MS = 1000;
    public static final int DEFAULT_MAX_QUEUE_SIZE = 100;

    private final Sender sender;
    private final int closeEnqueueTimeout;

    @ToString.Exclude private final BlockingQueue<Command> commandQueue;
    @ToString.Exclude private final Timer flushTimer;
    @ToString.Exclude private final Thread queueProcessorThread;
    @ToString.Exclude private final QueueProcessor queueProcessor;
    @ToString.Exclude private final Metrics metrics;
    @ToString.Exclude private final List<String> skipPatterns;

    private EnhancedRemoteReporter(Sender sender, int flushInterval, int maxQueueSize, int closeEnqueueTimeout,
                           Metrics metrics, List<String> skipPatterns) {
        this.sender = sender;
        this.metrics = metrics;
        this.closeEnqueueTimeout = closeEnqueueTimeout;
        this.skipPatterns = skipPatterns;
        commandQueue = new ArrayBlockingQueue<>(maxQueueSize);

        // start a thread to append spans
        queueProcessor = new QueueProcessor();
        queueProcessorThread = new Thread(queueProcessor, "jaeger.RemoteReporter-QueueProcessor");
        queueProcessorThread.setDaemon(true);
        queueProcessorThread.start();

        flushTimer = new Timer("jaeger.RemoteReporter-FlushTimer", true /* isDaemon */);
        flushTimer.schedule(
                new TimerTask() {
                    @Override
                    public void run() {
                        flush();
                    }
                },
                flushInterval,
                flushInterval);
    }

    @Override
    public void report(JaegerSpan span) {
        // skip report the spans matched by spel patterns
        if ( skipReport(span) ) {
            return;
        }

        // Its better to drop spans, than to block here
        boolean added = commandQueue.offer(new AppendCommand(span));

        if (!added) {
            metrics.reporterDropped.inc(1);
        }
    }

    private boolean skipReport(JaegerSpan span) {
        if (!CollectionUtils.isEmpty(skipPatterns) && span != null) {
            for ( String spel : skipPatterns ) {
                boolean matchResult = SpELUtil.matchRule(span, spel);
                if (matchResult) {
                    return Boolean.TRUE;
                }
            }
        }
        return Boolean.FALSE;
    }

    @Override
    public void close() {
        try {
            // best-effort: if we can't add CloseCommand in this time then it probably will never happen
            boolean added = commandQueue
                    .offer(new CloseCommand(), closeEnqueueTimeout, TimeUnit.MILLISECONDS);
            if (added) {
                queueProcessorThread.join(10000);
            } else {
                log.warn("Unable to cleanly close RemoteReporter, command queue is full - probably the"
                        + " sender is stuck");
            }
        } catch (InterruptedException e) {
            log.error("Interrupted", e);
            Thread.currentThread().interrupt();
        } finally {
            try {
                int n = sender.close();
                metrics.reporterSuccess.inc(n);
            } catch (SenderException e) {
                metrics.reporterFailure.inc(e.getDroppedSpanCount());
                log.error("Remote reporter error", e);
            }
            flushTimer.cancel();
        }
    }

    void flush() {
        // to reduce the number of updateGauge stats, we only emit queue length on flush
        metrics.reporterQueueLength.update(commandQueue.size());

        // We can safely drop FlushCommand when the queue is full - sender should take care of flushing
        // in such case
        commandQueue.offer(new FlushCommand());
    }

    /*
     * The code below implements the command pattern.  This pattern is useful for
     * situations where multiple threads would need to synchronize on a resource,
     * but are fine with executing sequentially.  The advantage is simplified code where
     * tasks are put onto a blocking queue and processed sequentially by another thread.
     */
    public interface Command {
        void execute() throws SenderException;
    }

    class AppendCommand implements Command {
        private final JaegerSpan span;

        public AppendCommand(JaegerSpan span) {
            this.span = span;
        }

        @Override
        public void execute() throws SenderException {
            int n = sender.append(span);
            if (n > 0) {
                metrics.reporterSuccess.inc(n);
            }
        }
    }

    class CloseCommand implements Command {
        @Override
        public void execute() throws SenderException {
            queueProcessor.close();
        }
    }

    class FlushCommand implements Command {
        @Override
        public void execute() throws SenderException {
            int n = sender.flush();
            metrics.reporterSuccess.inc(n);
        }
    }

    /*
     * This class creates a Runnable that is responsible for appending spans using a sender.
     */
    @ToString
    class QueueProcessor implements Runnable {
        private boolean open = true;

        @Override
        public void run() {
            while (open) {
                try {
                    Command command = commandQueue.take();

                    try {
                        command.execute();
                    } catch (SenderException e) {
                        metrics.reporterFailure.inc(e.getDroppedSpanCount());
                    }
                } catch (Exception e) {
                    log.error("QueueProcessor error:", e);
                    // Do nothing, and try again on next span.
                }
            }
        }

        public void close() {
            open = false;
        }
    }

    public static class Builder {
        private Sender sender;
        private int flushInterval = DEFAULT_FLUSH_INTERVAL_MS;
        private int maxQueueSize = DEFAULT_MAX_QUEUE_SIZE;
        private int closeEnqueTimeout = DEFAULT_CLOSE_ENQUEUE_TIMEOUT_MILLIS;
        private Metrics metrics;
        private List<String> skipPatterns;

        public Builder withFlushInterval(int flushInterval) {
            this.flushInterval = flushInterval;
            return this;
        }

        public Builder withMaxQueueSize(int maxQueueSize) {
            this.maxQueueSize = maxQueueSize;
            return this;
        }

        public Builder withMetrics(Metrics metrics) {
            this.metrics = metrics;
            return this;
        }

        public Builder withSender(Sender sender) {
            this.sender = sender;
            return this;
        }

        public Builder withCloseEnqueueTimeout(int closeEnqueueTimeoutMs) {
            this.closeEnqueTimeout = closeEnqueueTimeoutMs;
            return this;
        }

        public Builder withSkipPatterns(List<String> skipPatterns) {
            this.skipPatterns = skipPatterns;
            return this;
        }

        public EnhancedRemoteReporter build() {
            if (sender == null) {
                sender = SenderResolver.resolve();
            }
            if (metrics == null) {
                metrics = new Metrics(new InMemoryMetricsFactory());
            }
            return new EnhancedRemoteReporter(sender, flushInterval, maxQueueSize, closeEnqueTimeout, metrics, skipPatterns);
        }
    }
}
